package cc.toprank.syncdatatask.canal;

import cc.toprank.syncdatatask.util.PropFileHelper;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
/** 
 * description bin-log 配置
 * User: WeiDa
 * Date: 2017-5-7
 * Time: 16:56
 */
@Component
public class CanalClient extends AbstractCanalClient {


    public CanalClient(){
        super();
    }

    public CanalClient(String destination) {
        super(destination);
    }

    public static void startCanal(RabbitTemplate rabbitTemplate)  {
        // 根据ip，直接创建链接，无HA的功能AddressUtils.getHostIp()
        String destination = "example";
        String canalServerPort =  PropFileHelper.getInstance().getCanalConfig("canal.server.port");
        String canalInstanceName =  PropFileHelper.getInstance().getCanalConfig("canal.instance.name");
        String canalServerIp = PropFileHelper.getInstance().getCanalConfig("canal.server.ip");
        String dbName = PropFileHelper.getInstance().getCanalConfig("db.name");
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalServerIp, Integer.parseInt(canalServerPort)), canalInstanceName,
                "", "");

        final CanalClient canalClient = new CanalClient(destination);
        canalClient.setConnector(connector);
        canalClient.setRabbitTemplate(rabbitTemplate);
        canalClient.setDbName(dbName);
        canalClient.start();
        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal client");
                    canalClient.stop();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e));
                } finally {
                    logger.info("## canal client is down.");
                }
            }

        });
    }

}
